package com.yupi.moonBI.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

import java.util.Scanner;

/**
 * 死信队列
 */
public class DlxdirectProducer {

    private static final String DEAD_EXCHANGE_NAME = "dlx_direct_exchange";

    private static final String WORK_EXCHANGE_NAME = "direct2_exchange";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.88.130");
        factory.setPort(5672);
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            //声明死信交换机
            channel.exchangeDeclare(DEAD_EXCHANGE_NAME, "direct");

            String queueName = "boss_dlx_queue";
            channel.queueDeclare(queueName, true, false, false, null);
            channel.queueBind(queueName, DEAD_EXCHANGE_NAME, "boss");

            String queueName02 = "manage_dlx_queue";
            channel.queueDeclare(queueName02, true, false, false, null);
            channel.queueBind(queueName02, DEAD_EXCHANGE_NAME, "manage");

            DeliverCallback deliverCallback1 = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");
                // 拒绝消息
                channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false,false);
                System.out.println(" [boss] Received '" +
                        delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
            };

            DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");
                // 拒绝消息
                channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false,false);
                System.out.println(" [manage] Received '" +
                        delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
            };

            channel.basicConsume(queueName, false, deliverCallback1, consumerTag -> {
            });
            channel.basicConsume(queueName02, false, deliverCallback2, consumerTag -> {
            });

            Scanner scanner = new Scanner(System.in);
            while (scanner.hasNext()) {
                String userInput = scanner.nextLine();
                String[] strings = userInput.split(" ");
                if (strings.length<1){
                    continue;
                }
                String message = strings[0];
                String routingKey = strings[1];
                //设置消息过期时间，模拟死信
                /*AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                        .expiration("10000")
                        .build();//设置消息的过期时间，单位是ms
                channel.basicPublish(WORK_EXCHANGE_NAME, routingKey, properties, message.getBytes("UTF-8"));*/
                channel.basicPublish(WORK_EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
                System.out.println(" [x] Sent '" + message + "with routing:"+routingKey+"'");
            }

        }
    }
}
